package com.group1.realtime.app.dwd.db;

import com.group1.realtime.app.BaseSQLApp;
import com.group1.realtime.common.Constant;
import com.group1.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DWD_UserNewAdd extends BaseSQLApp {

    public static void main(String[] args) {
        new DWD_UserNewAdd().init(
                2022,
                2,
                "DWD_UserNewAdd"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        readOdsDb(tEnv, "DWD_UserNewAdd");

        //从ods_db中过滤出加购数据
        Table userInfo = tEnv.sqlQuery("select " +
                " `data`['id'] id, " +
                " `data`['login_name'] login_name, " +
                " `data`['email'] email, " +
                " `data`['user_level'] user_level, " +
                " `data`['birthday'] birthday, " +
                " `data`['gender'] gender, " +
                " ts " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='user_info' " +
                "and `type`='insert' " );


        tEnv.executeSql("create table dwd_user_new_add(" +
                "id string," +
                "login_name string," +
                "email string," +
                "user_level string," +
                "birthday  String," +
                "gender String," +
                "ts String" +
                ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_USER_NEW_ADD));

        userInfo.executeInsert("dwd_user_new_add");

    }
}
